package com.atguigu.flink.sql;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

/**
 * Created by Smexy on 2023/2/6
 *
 *  TVFwindow 对比 Group Window:
 *          ①提供性能优化措施
 *          ②支持 GroupingSets 语法
 *                  select xx from  xx group by a,b
 *                  grouping sets ( (a,b), (a) , (b) , () )  hive,flink支持
 *                  cube      ck,hive,flink支持
 *                  rollup     ck,hive,flink支持
 *                  withTotal   只有ck支持
 *          ③支持窗口中求TopN
 *
 *    缺陷: 目前1.13仅支持
 *      Tumble Windows
 *      Hop Windows
 *      Cumulate Windows   累积窗口
 *      Session Windows (will be supported soon)  目前不支持。
 */
public class Demo12_TVFWindowGroupingSets
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        env.setParallelism(1);

        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
            .<WaterSensor>forMonotonousTimestamps()
            .withTimestampAssigner((e, ts) -> e.getTs());


        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction())
            .assignTimestampsAndWatermarks(watermarkStrategy);


        /*
                只选取流中数据的某些属性，组成Table
         */
        Table table = tableEnvironment.fromDataStream(ds, $("id"),$("ts"),$("vc"),
            $("pt").proctime(),$("et").rowtime()
        );

        tableEnvironment.createTemporaryView("t1",table);

        /*
                window_start,window_end 是关键字，代表窗口的其实和终止的时间范围

                声明滚动窗口:  FROM TABLE (
                        TUMBLE(TABLE 表名, DESCRIPTOR(时间字段), INTERVAL '10' MINUTES  slide|size)
                  )
         */
        String tumble = "SELECT window_start, window_end, id, SUM(vc) sumVC" +
            "  FROM TABLE(" +
            "    TUMBLE(TABLE t1, DESCRIPTOR(et), INTERVAL '5' SECONDS))" +
            "  GROUP BY window_start, window_end , " +
            "  rollup(id) ";
            //"  CUBE( id ) ";
            //"  GROUPING SETS ( (id),() ) " ;



        //不支持session
        tableEnvironment.sqlQuery(tumble)
                        .execute()
                        .print();




    }
}
